package com.moli.iov.down.command.mq.kafka;

import com.moli.iov.down.command.mq.kafka.handle.IovLoginHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

/**
 * <p>
 *  kafka 消费者  上行报文 登陆
 * </p>
 *
 * @author: zql
 * @Date: 2018/11/30 14:40
 */
public class ConsumerMessageLoginListener implements AcknowledgingMessageListener<String,Object> {

    protected final Logger log = LoggerFactory.getLogger(ConsumerMessageLoginListener.class);

    @Autowired
    private TaskExecutor taskExecutor;

    @Autowired
    private IovLoginHandler iovLoginHandler;

    @Override
    public void onMessage(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment) {
        try {
            taskExecutor.execute(()->{
                String key = record.key();
                try {
                    iovLoginHandler.responseHandle(key);
                }catch(Exception e){
                    log.error("afka Consumer [上行报文]  登陆 线程池中处理异常  vin: {}", key, e);
                }
            });
        }catch(Exception e){
            log.error("kafka 消费者   [上行报文]  登陆 添加线程池异常", e);
        }finally {
            //手动提交
            acknowledgment.acknowledge();
        }
    }
}
